package com.atguigu.utils;

import com.alibaba.fastjson.JSONObject;
import org.apache.flink.api.common.serialization.DeserializationSchema;
import org.apache.flink.api.common.serialization.SimpleStringSchema;
import org.apache.flink.api.common.typeinfo.BasicTypeInfo;
import org.apache.flink.api.common.typeinfo.TypeInformation;
import org.apache.flink.connector.kafka.source.KafkaSource;
import org.apache.flink.connector.kafka.source.enumerator.initializer.OffsetsInitializer;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaProducer;
import org.apache.flink.streaming.connectors.kafka.KafkaSerializationSchema;
import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.clients.producer.ProducerRecord;

import javax.annotation.Nullable;
import java.io.IOException;
import java.util.Properties;

/**
 * ClassName: KafkaUtil
 * Package: com.atguigu.utils
 * Description:
 *
 * @Author Lovxy
 * @Create 2023/5/15 10:48
 * @Version 1.0
 */
public class KafkaUtil {
    private static final String BOOTSTRAP_SERVER = "hadoop102:9092";

    public static KafkaSource<String> getKafkaSource(String topic, String groupId) {
        return KafkaSource.<String>builder()
                .setBootstrapServers(BOOTSTRAP_SERVER)
                .setGroupId(groupId)
                .setTopics(topic)
                .setValueOnlyDeserializer(new DeserializationSchema<String>() {
                    @Override
                    public String deserialize(byte[] message) throws IOException {
                        if (message != null) {
                            return new String(message);
                        } else {
                            return null;
                        }
                    }

                    @Override
                    public boolean isEndOfStream(String nextElement) {
                        return false;
                    }

                    @Override
                    public TypeInformation<String> getProducedType() {
                        return BasicTypeInfo.STRING_TYPE_INFO;
                    }
                })
                .setStartingOffsets(OffsetsInitializer.latest())
                .build();
    }

    public static FlinkKafkaProducer<String> getFlinkKafkaProducer(String topic) {
        return new FlinkKafkaProducer<String>(BOOTSTRAP_SERVER,
                topic,
                new SimpleStringSchema());
    }

    public static FlinkKafkaProducer<JSONObject> getFlinkKafkaProducer() {
        Properties properties = new Properties();
        properties.setProperty(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, BOOTSTRAP_SERVER);
        return new FlinkKafkaProducer<JSONObject>("dwd_default_topic",
                new KafkaSerializationSchema<JSONObject>() {
                    @Override
                    public ProducerRecord<byte[], byte[]> serialize(JSONObject element, @Nullable Long timestamp) {
                        return new ProducerRecord<>(element.getString("sink_table"),
                                element.getString("data").getBytes());
                    }
                }, properties, FlinkKafkaProducer.Semantic.EXACTLY_ONCE);
    }

    public static <T> FlinkKafkaProducer<T> getFlinkKafkaProducer(KafkaSerializationSchema<T> kafkaSerializationSchema) {
        Properties properties = new Properties();
        properties.setProperty(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, BOOTSTRAP_SERVER);
        return new FlinkKafkaProducer<T>(
                "dwd_default_topic",
                kafkaSerializationSchema,
                properties,
                FlinkKafkaProducer.Semantic.EXACTLY_ONCE);
    }

    public static String getTopicDbDDL(String groupId) {
        return "create table topic_db( " +
                "    `database` String, " +
                "    `table` String, " +
                "    `type` String, " +
                "    `data` Map<String,String>, " +
                "    `old` Map<String,String>, " +
                "    `pt` as proctime() " +
                ") " + getKafkaSourceConnOption("topic_db", groupId);
    }
    public static String getTopicLogDDL(String groupId) {
        return "create table topic_log(common MAP<STRING,STRING> COMMENT '公共信息',\n" +
                " `page` MAP<STRING,STRING> COMMENT '页面信息',\n" +
                " `actions` ARRAY<MAP<STRING,STRING>> COMMENT '动作信息',\n" +
                " `displays` ARRAY<MAP<STRING,STRING>> COMMENT '曝光信息',\n" +
                " `start` MAP<STRING,STRING> COMMENT '启动信息',\n" +
                " `err` MAP<STRING,STRING> COMMENT '错误信息',\n" +
                " `appVideo` MAP<STRING,STRING> COMMENT '播放日志',\n" +
                " ts BIGINT COMMENT '时间戳') " + getKafkaSourceConnOption("topic_log", groupId);
    }

    public static String getKafkaSourceConnOption(String topic, String groupId) {
        return "with ( " +
                "    'connector' = 'kafka', " +
                "    'topic' = '" + topic + "', " +
                "    'properties.bootstrap.servers' = 'hadoop102:9092', " +
                "    'properties.group.id' = '" + groupId + "', " +
                "    'scan.startup.mode' = 'latest-offset', " +
                "    'format' = 'json' " +
                ")";
    }

    public static String getKafkaSinkConnOption(String topic) {
        return "with ( " +
                "    'connector' = 'kafka', " +
                "    'topic' = '" + topic + "', " +
                "    'properties.bootstrap.servers' = '" + BOOTSTRAP_SERVER + "', " +
                "    'format' = 'json' " +
                ")";
    }

    public static String getKafkaUpsertSinkConnOption(String topic) {
        return "WITH ( " +
                "  'connector' = 'upsert-kafka', " +
                "  'topic' = '" + topic + "', " +
                "  'properties.bootstrap.servers' = '" + BOOTSTRAP_SERVER + "', " +
                "  'key.format' = 'json', " +
                "  'value.format' = 'json' " +
                ")";
    }
}
